package com.mini.cn.goods.servlet.ws.service.impl;


import com.mini.cn.goods.io.netty.ws.WebSocketSession;
import com.mini.cn.goods.io.netty.ws.WsConfig;
import com.mini.cn.goods.io.netty.ws.info.FastPushMsg;
import com.mini.cn.goods.io.netty.ws.info.LCDataPacket;
import com.mini.cn.goods.io.netty.ws.info.PushMessage;
import com.mini.cn.goods.io.netty.ws.info.SessionUserInfo;
import com.mini.cn.goods.servlet.ws.WSApplicationData;
import com.mini.cn.goods.servlet.ws.common.PushMessageFactory;
import com.mini.cn.goods.servlet.ws.serializer.ISerializer;
import com.mini.cn.goods.servlet.ws.serializer.SerializerFactory;
import com.mini.cn.goods.servlet.ws.service.IPushResultCallback;
import com.mini.cn.goods.servlet.ws.service.IPushService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * User: fly
 * Date: 2020-02-12
 * Time: 6:34 下午
 */
@Slf4j
@Service
public class PushServiceImpl implements IPushService {

    @Override
    public void pushAllUser(PushMessage pushMessage, IPushResultCallback callback) {
        String msgSceneId = pushMessage.getMsgSceneId();
        ConcurrentHashMap<WebSocketSession, Boolean> socketSessions = WSApplicationData.INSTANCE.getRegisterMsgMap().get(msgSceneId);
        if (socketSessions == null || socketSessions.isEmpty()) {
            return;
        }
        ISerializer serializer = SerializerFactory.INTANCE.getIntanceSerializer(ISerializer.JSON);
        if (serializer == null) {
            return;
        }
        Iterator<WebSocketSession> iterator = socketSessions.keySet().iterator();
        while (iterator.hasNext()) {
            WebSocketSession session = iterator.next();
            if (session == null || !session.isActive()) {
                continue;
            }
            if (!session.isWritable()) {
                continue;
            }
            SessionUserInfo userInfo = session.getAttribute(WebSocketSession.USERKEY);
            userInfo.getWriteSeqId().increment();
            pushMessage.setSeqId(userInfo.getWriteSeqId().intValue());
            pushMessage.setTimestamp(System.currentTimeMillis());
            byte[] content = serializer.serializeBytes(pushMessage);
            //推送消息
            ByteBuf byteBuf = LCDataPacket.coverToBytesNoPool(content);
            //是否分包的标示
            boolean isUnpack = false;
            //分包
            List<ByteBuf> byteBufs = null;
            if (byteBuf.capacity() > WsConfig.UNPACKSIZE) {
                byteBufs = unpageParkageByteBuf(byteBuf);
                isUnpack = true;
            }
            //做发送的try
            try {
                if (!isUnpack) {
                    session.sendBinary(byteBuf).addListener(future -> {
                        futureCallBack(future, callback, session, pushMessage);
                    });
                } else {
                    ChannelFuture channelFuture = session.sendUnpackBinaryByByteBufs(byteBufs, true);
                    if (channelFuture != null) {
                        channelFuture.addListener(future -> {
                            futureCallBack(future, callback, session, pushMessage);
                        });
                    }
                }
            } catch (Exception e) {

            }
        }
    }


    @Override
    public void pushAllUserFast(FastPushMsg fastPushMsg, IPushResultCallback callback) {
        //添加cat 监控
        String msgId = fastPushMsg.getMsgSceneId();
        LCDataPacket out = PushMessageFactory.coverInToOuT(fastPushMsg);
        ByteBuf byteBuf = LCDataPacket.coverToBytes(out);
        BinaryWebSocketFrame frame = new BinaryWebSocketFrame(byteBuf);

        ConcurrentHashMap<WebSocketSession, Boolean> list = WSApplicationData.INSTANCE.
                getRegisterMsgMap().get(msgId);
        if (list == null || list.isEmpty()) {
            return;
        }
        Iterator<WebSocketSession> iterator = list.keySet().iterator();
        while (iterator.hasNext()) {
            WebSocketSession socketSession = iterator.next();
            if (socketSession == null || !socketSession.isActive()) {
                continue;
            }
            if (!socketSession.isWritable()) {
                continue;
            }
            try {
                BinaryWebSocketFrame newFrame = frame.duplicate().retain();
                //多次发送防止被realse
                socketSession.sendBinary(newFrame).addListener(future -> {
                    futureCallBack(future, callback, socketSession, fastPushMsg);
                });
            } catch (Exception e) {

            }
        }
    }

    @Override
    public void pushAllUserFastUnpack(FastPushMsg fastPushMsg, IPushResultCallback callback) {
        String msgId = fastPushMsg.getMsgSceneId();
        LCDataPacket out = PushMessageFactory.coverInToOuT(fastPushMsg);
        ByteBuf byteBuf = LCDataPacket.coverToBytes(out);
        //分包
        boolean isUnpack = false;
        List<ByteBuf> byteBufs = null;
        if (byteBuf.capacity() > WsConfig.UNPACKSIZE) {
            byteBufs = unpageParkageByteBuf(byteBuf);
            isUnpack = true;
        }
        BinaryWebSocketFrame frame = null;
        if (!isUnpack) {
            frame = new BinaryWebSocketFrame(byteBuf);
        }
        try {
            ConcurrentHashMap<WebSocketSession, Boolean> list = WSApplicationData.INSTANCE.
                    getRegisterMsgMap().get(msgId);
            if (list == null || list.isEmpty()) {
                return;
            }
            Iterator<WebSocketSession> iterator = list.keySet().iterator();
            while (iterator.hasNext()) {
                WebSocketSession socketSession = iterator.next();
                if (socketSession == null || !socketSession.isActive()) {
                    continue;
                }
                if (!socketSession.isWritable()) {
                    continue;
                }
                try {
                    if (!isUnpack) {
                        BinaryWebSocketFrame newFrame = frame.duplicate().retain();
                        //多次发送防止被realse
                        socketSession.sendBinary(newFrame).addListener(future -> {
                            futureCallBack(future, callback, socketSession, fastPushMsg);
                        });
                    } else {
                        ChannelFuture channelFuture = socketSession.
                                sendUnpackBinaryByByteBufs(byteBufs, false);
                        if (channelFuture != null) {
                            channelFuture.addListener(future -> {
                                futureCallBack(future, callback, socketSession, fastPushMsg);
                            });
                        }
                    }
                } catch (Exception e) {

                }
            }
        } finally {
            if (isUnpack) {
                //分包
                for (ByteBuf buf : byteBufs) {
                    buf.release();
                }
            } else {
                //没有分包
                byteBuf.release();
            }
        }
    }

    @Override
    public void pushUserByUserId(PushMessage message, IPushResultCallback callback) {

        String uid = message.getTo();
        WebSocketSession session = WSApplicationData.INSTANCE.getUserSessionMap().get(uid);
        if (session == null || !session.isActive()) {
            return;
        }
        if (!session.isWritable()) {
            return;
        }
        SessionUserInfo info = session.getAttribute(WebSocketSession.USERKEY);
        info.getWriteSeqId().increment();
        message.setSeqId(info.getWriteSeqId().intValue());
        message.setTimestamp(System.currentTimeMillis());
        LCDataPacket duplexServerProtocal = PushMessageFactory.coverInToOuT(message);
        //推送消息
        ByteBuf byteBuf = LCDataPacket.coverToBytes(duplexServerProtocal);
        //分包
        List<ByteBuf> byteBufs = null;
        //是否分包的标示
        boolean isUnpack = false;
        if (byteBuf.capacity() > WsConfig.UNPACKSIZE) {
            //加入缓存策略
            byteBufs = unpageParkageByteBuf(byteBuf);
            isUnpack = true;
        }
        if (!isUnpack) {
            session.sendBinary(byteBuf).addListener(future -> {
                futureCallBack(future, callback, session, message);
            });
        } else {
            ChannelFuture channelFuture = session.sendUnpackBinaryByByteBufs(byteBufs, true);
            if (channelFuture != null) {
                channelFuture.addListener(future -> {
                    futureCallBack(future, callback, session, message);
                });
            }
        }
    }


    @Override
    public void pushUserPushMessage(PushMessage message, WebSocketSession session, IPushResultCallback callback) {

        if (session == null || !session.isActive()) {
            return;
        }
        if (!session.isWritable()) {
            return;
        }
        int seq = 1;
        SessionUserInfo userInfo = session.getAttribute(WebSocketSession.USERKEY);
        //判断是否注册 如果为null 则为注册失败的用户
        if (userInfo != null) {
            //设置seq
            userInfo.getWriteSeqId().increment();
            seq = userInfo.getWriteSeqId().intValue();
        }
        message.setSeqId(seq);
        //设置时间
        message.setTimestamp(System.currentTimeMillis());
        LCDataPacket duplexServerProtocal = PushMessageFactory.coverInToOuT(message);
        //推送消息
        ByteBuf byteBuf = LCDataPacket.coverToBytes(duplexServerProtocal);
        //分包
        List<ByteBuf> byteBufs = null;
        //是否分包的标示
        boolean isUnpack = false;
        if (byteBuf.capacity() > WsConfig.UNPACKSIZE) {
            byteBufs = unpageParkageByteBuf(byteBuf);
            isUnpack = true;
        }
        if (!isUnpack) {
            session.sendBinary(byteBuf).addListener(future -> {
                futureCallBack(future, callback, session, message);
            });
        } else {
            ChannelFuture channelFuture = session.sendUnpackBinaryByByteBufs(byteBufs, true);
            if (channelFuture != null) {
                channelFuture.addListener(future -> {
                    futureCallBack(future, callback, session, message);
                });
            }
        }
    }

    @Override
    public void pushUserPushMessageList(List<PushMessage> messages,
                                        WebSocketSession session, IPushResultCallback callback) {

        if (session == null || !session.isActive()) {
            return;
        }
        if (!session.isWritable()) {
            return;
        }
        //为socket的推送消息标记
        SessionUserInfo userInfo = session.getAttribute(WebSocketSession.USERKEY);
        for (PushMessage message : messages) {
            LCDataPacket duplexServerProtocal = PushMessageFactory.coverInToOuT(message);
            //推送消息
            ByteBuf byteBuf = LCDataPacket.coverToBytes(duplexServerProtocal);
            //分包
            List<ByteBuf> byteBufs = null;
            //是否分包的标示
            boolean isUnpack = false;
            if (byteBuf.capacity() > WsConfig.UNPACKSIZE) {
                byteBufs = unpageParkageByteBuf(byteBuf);
                isUnpack = true;
            }
            //设置seq
            userInfo.getWriteSeqId().increment();
            message.setSeqId(userInfo.getWriteSeqId().intValue());
            message.setTimestamp(System.currentTimeMillis());
            if (!isUnpack) {
                session.sendBinary(byteBuf).addListener(future -> {
                    futureCallBack(future, callback, session, message);
                });
            } else {
                ChannelFuture channelFuture = session.sendUnpackBinaryByByteBufs(byteBufs, true);
                if (channelFuture != null) {
                    channelFuture.addListener(future -> {
                        futureCallBack(future, callback, session, message);
                    });
                }
            }
        }
    }


    /**
     * 分包代码
     *
     * @param conent
     * @return
     */
    private List<byte[]> unpageParkage(String conent) {
        List<byte[]> bytes = new ArrayList<byte[]>();
        byte[] orbytes = conent.getBytes();
        int len = orbytes.length;
        //大于200k 分包
        int size = len / WsConfig.UNPACKSIZE;
        int mode = len % WsConfig.UNPACKSIZE;
        for (int i = 0; i < size; i++) {
            byte[] b = new byte[WsConfig.UNPACKSIZE];
            int from = WsConfig.UNPACKSIZE * i;
            int to = WsConfig.UNPACKSIZE * (i + 1);
            b = Arrays.copyOfRange(orbytes, from, to);
            bytes.add(b);
        }
        if (mode != 0) {
            byte[] leftb = new byte[mode];
            leftb = Arrays.copyOfRange(orbytes, size * WsConfig.UNPACKSIZE,
                    len);
            bytes.add(leftb);
        }
        return bytes;
    }

    /**
     * 调用者记住释放
     *
     * @param bytes
     * @return
     */
    private List<ByteBuf> allocByteBufByBytes(List<byte[]> bytes) {
        List<ByteBuf> byteBufs = new ArrayList<ByteBuf>();
        for (byte[] b : bytes) {
            //有多个包
            ByteBuf buffer = Unpooled.directBuffer(b.length);
            buffer.writeBytes(b);
            byteBufs.add(buffer);
        }
        return byteBufs;
    }

    /**
     * 分包发送
     *
     * @param buf
     * @return
     */
    private List<ByteBuf> unpageParkageByteBuf(ByteBuf buf) {
        List<ByteBuf> byteBufs = new ArrayList<ByteBuf>();
        int len = buf.capacity();
        //大于200k 分包
        int size = len / WsConfig.UNPACKSIZE;
        int mode = len % WsConfig.UNPACKSIZE;
        for (int i = 0; i < size; i++) {
            //有多个包
            ByteBuf buffer = Unpooled.directBuffer(WsConfig.UNPACKSIZE);
            buf.readBytes(buffer);
            byteBufs.add(buffer);
        }
        if (mode != 0) {
            ByteBuf leftBuffer = Unpooled.directBuffer(mode);
            buf.readBytes(leftBuffer);
            byteBufs.add(leftBuffer);
        }
        //释放原包
        ReferenceCountUtil.release(buf);
        return byteBufs;
    }


    /**
     * 释放
     *
     * @param byteBufs
     */
    private void releaseByteBuff(List<ByteBuf> byteBufs) {
        for (ByteBuf buf : byteBufs) {
            //防止内存泄漏
            if (buf.refCnt() > 0) {
                ReferenceCountUtil.release(buf);
            }
        }
        byteBufs.clear();
        byteBufs = null;
    }


    /**
     * future 统一回调代码
     *
     * @param future   异步回调
     * @param callback
     * @param session
     * @param message
     */
    private void futureCallBack(Future future, IPushResultCallback callback,
                                WebSocketSession session, FastPushMsg message) {
        if (callback != null) {
            if (future.isSuccess()) {
                callback.onSuccessSend(session, message);
            } else {
                callback.onFailSend(session, message, future.cause());
            }
        }
    }
}
